package org.raymond.iworks.study.mq.rabbitmq.stream.official;

import com.rabbitmq.stream.Address;
import com.rabbitmq.stream.Environment;

import java.util.Scanner;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.IntStream;

/**
 * @author raymond
 * @version V1.0
 * @Description: TODO
 */
public class SingleActiveConsumer {
    public static void main(String[] args) {
        try (Environment environment =
                     Environment.builder()
                             .host("192.168.0.125")
                             .port(5552)
                             .username("cloud")
                             .password("cloud123")
                             .virtualHost("/cloud")
                             .addressResolver(address -> {
                                 System.out.println("address: "+address.host().toString());
                                 Address entryPoint = new Address("192.168.0.125", 5552);
                                 return entryPoint;
                             })
                             .maxConsumersByConnection(1)
                             .maxTrackingConsumersByConnection(1)
                             .build()) {

            String stream = "single-active-consumer";
            String reference = "my-app";
            environment.streamCreator().stream(stream).create();
            System.out.println("Created stream " + stream);

            AtomicLong sequence = new AtomicLong(0);
            IntStream.range(0, 3)
                    .forEach(
                            i -> {
                                System.out.println("Starting consumer instance " + i);
                                environment.consumerBuilder().stream(stream)
                                        .name(reference)
                                        .singleActiveConsumer()
                                        .autoTrackingStrategy()
                                        .messageCountBeforeStorage(10)
                                        .builder()
                                        .messageHandler(
                                                (context, message) -> {
                                                    System.out.printf(
                                                            "Consumer instance %d received a message (%d).%n",
                                                            i, sequence.incrementAndGet()
                                                    );
                                                })
                                        .build();
                            });
            Scanner keyboard = new Scanner(System.in);
            keyboard.nextLine();
        }
    }
}
